package org.example.flink.triger;

import cn.hutool.core.date.DateUtil;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.example.flink.eventtime.CustomSource;
import org.example.flink.eventtime.S0;

import java.util.Iterator;

/**
 * @author zhangjg
 * @desc x
 * @date 2022/08/08 16:40
 **/
public class Test {

    public static void main(String[] args) throws Exception {

        Configuration cfg = new Configuration();
//        cfg.setString("execution.savepoint.path","file:///D:/data/ck/348ef2976ea26f2b1a6e3b813b4b2939/chk-25");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(cfg);
        env.enableCheckpointing(1000000);
        env.getCheckpointConfig().setCheckpointStorage("file:///D:/data/ck/");
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(5);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000));
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.setParallelism(1);

        DataStreamSource<S0> source = env.addSource(new CustomSource2());
        source.name("自定义数据源");
        source.setParallelism(1);
//        SingleOutputStreamOperator<S0> s1 = source.assignTimestampsAndWatermarks(WatermarkStrategy.<S0>forBoundedOutOfOrderness(Duration.ofSeconds(5))
//                .withTimestampAssigner((e,time) -> DateUtil.parse(e.getEventtime(),"yyyy-MM-dd HH:mm:ss").getTime()));

        source.keyBy(x -> x.getName())
//                .window(TumblingEventTimeWindows.of(Time.seconds(60)))
                .window(TumblingProcessingTimeWindows.of(Time.seconds(60)))
                .trigger(CountAndPressTimeTrigger.of(10,Time.seconds(10)))
                .process(new ProcessWindowFunction<S0, Tuple4<String,String,String,Integer>, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<S0> elements, Collector<Tuple4<String,String,String, Integer>> out) throws Exception {
                        Iterator<S0> iterator = elements.iterator();
                        int cnt = 0;
                        while (iterator.hasNext()){
                            iterator.next();
                            cnt++;
                        }
                        out.collect(Tuple4.of(DateUtil.date(context.window().getStart()).toString("yyyy-MM-dd HH:mm:ss"),DateUtil.date(context.window().getEnd()).toString("yyyy-MM-dd HH:mm:ss"),s,cnt));
                    }
                }).addSink(new SinkFunction<Tuple4<String, String, String, Integer>>() {
            @Override
            public void invoke(Tuple4<String, String, String, Integer> value, Context context) throws Exception {
//                System.out.println(value.toString());
            }
        }).name("sink").setParallelism(1);

        env.execute("EventTime Test");

    }

}
